package com.log.consume.config;

import com.log.consume.collect.KafkaLogCollect;
import com.log.consume.collect.RedisLogCollect;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.InputStream;
import java.util.Properties;

@Component
public class Start implements CommandLineRunner {
    private static org.slf4j.Logger logger = LoggerFactory.getLogger(Start.class);
    private Properties properties = new Properties();
    private String model;
    private String kafkaHosts;

    private String esHosts;
    private String esUserName;
    private String esPassWord;

    private String redisHost;
    private String redisPassWord;

    private String KAFKA_MODE_NAME = "kafka";
    private String REDIS_MODE_NAME = "redis";


    private void loadConfig() {
        try {
            InputStream in = KafkaLogCollect.class.getClassLoader().getResourceAsStream("plumelog.properties");
            this.properties.load(in);

            InitConfig.MAX_SEND_SIZE = Integer.valueOf(this.properties.getProperty("plumelog.server.maxSendSize"));
            InitConfig.KAFKA_GROUP_NAME = this.properties.getProperty("plumelog.server.kafka.kafkaGroupName");
            InitConfig.MAX_INTERVAL = Integer.valueOf(this.properties.getProperty("plumelog.server.interval"));

            this.kafkaHosts = this.properties.getProperty("plumelog.server.kafka.kafkaHosts");

            this.esHosts = this.properties.getProperty("plumelog.server.es.esHosts");
            this.esUserName = this.properties.getProperty("plumelog.server.es.userName");
            this.esPassWord = this.properties.getProperty("plumelog.server.es.passWord");

            this.redisHost = this.properties.getProperty("plumelog.server.redis.redisHost");
            this.redisPassWord = this.properties.getProperty("plumelog.server.redis.redisPassWord");

            this.model = this.properties.getProperty("plumelog.server.model");
        } catch (Exception e) {
            logger.error("load config fail!", e);
        }

    }

    private void serverStart() {
        loadConfig();
        logger.info("load config success!");
        if (KAFKA_MODE_NAME.equals(model)) {
            logger.info("kafkaHosts:" + kafkaHosts);
            logger.info("esHosts:" + esHosts);
            KafkaLogCollect kafkaLogCollect = new KafkaLogCollect(this.kafkaHosts, this.esHosts, this.esUserName, this.esPassWord);
            kafkaLogCollect.kafkaStart();
        }
        if (REDIS_MODE_NAME.equals(model)) {
            String[] hs = redisHost.split(":");
            String ip = hs[0];
            int port = Integer.valueOf(hs[1]);
            logger.info("redisHost:" + redisHost);
            logger.info("esHosts:" + esHosts);
            RedisLogCollect redisLogCollect = new RedisLogCollect(ip, port, this.redisPassWord, this.esHosts, this.esUserName, this.esPassWord);
            redisLogCollect.redisStart();
        }
    }


    @Override
    public void run(String... args) {
        try {
            logger.info("-------------------启动加载任务,收集日志--------------------------");
            Start start = new Start();
            start.serverStart();
        } catch (Exception e) {
            logger.error("plumelog server running fail!", e);
        }
    }
}
